package cn._51doit.flink.day12;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 按照各个维度统计新增用的信息
 */
public class TopNDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        tEnv.executeSql(
                "CREATE TABLE tb_user_order (\n" +
                        "  `uid`      STRING ,\n" +
                        "  `province` STRING ,\n" +
                        "  `city`     STRING ,\n" +
                        "  `money`    DOUBLE  \n" +
                        ") WITH (\n" +
                        "  'connector' = 'kafka',\n" +
                        "  'topic' = 'user-orders2',\n" +
                        "  'properties.bootstrap.servers' = 'node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092',\n" +
                        "  'properties.group.id' = 'testGroup',\n" +
                        "  'scan.startup.mode' = 'earliest-offset',\n" +
                        "  'format' = 'csv'\n" +
                        ")"
        );

        tEnv.executeSql(
                "CREATE TABLE tb_print_table (\n" +
                        "  `rk` BIGINT ,\n" +
                        "  `province` STRING ,\n" +
                        "  `city`    STRING ,\n" +
                        "  `total_money`    DOUBLE  \n" +
                        ") WITH (\n" +
                        "  'connector' = 'print'\n" +
                        ")"
        );

        tEnv.executeSql(
                "insert into tb_print_table select\n" +
                        "  rk,\n" +
                        "  province,\n" +
                        "  city,\n" +
                        "  total_money\n" +
                        "from\n" +
                        "(\n" +
                        "  select\n" +
                        "    province,\n" +
                        "    city,\n" +
                        "    total_money,\n" +
                        "    row_number() over(partition by province order by total_money desc) rk\n" +
                        "  from\n" +
                        "  (\n" +
                        "    select\n" +
                        "      province, city, sum(money) total_money\n" +
                        "    from tb_user_order\n" +
                        "      group by province, city\n" +
                        "    \n" +
                        "  )\n" +
                        ") where rk <= 2"
        );

    }
}
